package com.navinfo.tripanalysis.service.impl;

import com.navinfo.tripanalysis.common.arithmetic.common.VehicleInfo;
import com.navinfo.tripanalysis.service.AbstractLoadUreaInfoService;
import com.navinfo.tripanalysis.service.AbstractLoadVehicleInfoService;
import org.apache.commons.io.FileUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.io.File;
import java.io.IOException;
import java.net.URISyntaxException;
import java.util.*;

/**
 * 加载东风车辆的配置信息
 * @author web
 */
public class LoadUreaInfoServiceImpl_DF extends AbstractLoadUreaInfoService {

    @Override
    protected String getTable() {
        return  "(select value from urea_level_change where type = 1) t";
    }

    @Override
    protected JavaRDD<Long> getUreaInfoPair(SparkSession session, JavaRDD<Row> rowJavaRDD) {
//        Broadcast<Map<String, String>> broadcast = new JavaSparkContext(session.sparkContext()).broadcast(getWeightLoad());
        List<Long> ureaLevelList = new ArrayList<>();

        JavaRDD<Long> ureaInfoJavaRDD = rowJavaRDD.mapPartitions(rowIterator -> {

            Long ureaAbnormalLevel = 0L ;
            while (rowIterator.hasNext()) {
                Row row = rowIterator.next();
                ureaAbnormalLevel = Long.valueOf(row.getString(0));
                ureaLevelList.add(new Long(ureaAbnormalLevel));

            }
            return ureaLevelList.iterator();
        });
        return ureaInfoJavaRDD;
    }

    private Map<String, String> getWeightLoad(){
        Map<String, String> result = new HashMap<>();

        List<String> weightLoads = null;
        try {
            File file = FileUtils.getFile( new File(this.getClass().getProtectionDomain().getCodeSource().getLocation().toURI().getPath()).getParentFile(), "df_weight_load") ;
            weightLoads = FileUtils.readLines(file);
        } catch (URISyntaxException |IOException e) {
            logger.error("加载东风车型载重信息失败...", e);
        }

        if(null != weightLoads){
            weightLoads.forEach(e ->{
                int firstIndex = e.indexOf(",");
                result.put(e.substring(0,firstIndex),  e.substring(firstIndex+1));
            });
        }

        logger.info("加载东风车型载重信息完成，大小{}...",result.size());
        return result;
    }

}
